package io.gitee.tziye.core.kv.impl;

import io.gitee.tziye.common.RenaiUtil;
import io.gitee.tziye.common.ValueResult;
import io.gitee.tziye.core.instance.DefaultRenaiClient;
import io.gitee.tziye.core.kv.PriorityQOperator;
import io.gitee.tziye.core.kv.RenaiKey;
import io.gitee.tziye.core.kv.RenaiValue;
import io.gitee.tziye.enums.KvType;
import io.gitee.tziye.jdbc.JdbcProxy;
import io.gitee.tziye.jdbc.ProcedureResult;
import io.vavr.Tuple2;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;

import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.Map;

import static io.gitee.tziye.common.Constants.*;

public class DefaultPriorityQOperator extends PriorityQOperator {

    private final KvCommonOperator kvCommonOperator;
    private final JdbcProxy jdbcProxy;

    public DefaultPriorityQOperator(DefaultRenaiClient renaiClient) {
        super(renaiClient);
        kvCommonOperator = new KvCommonOperator(renaiClient);
        jdbcProxy = renaiClient.getJdbcProxy();
    }

    @Override
    public void saveKey(String key, Integer expireSeconds) {
        kvCommonOperator.saveKey(key, expireSeconds, KvType.PriorityQ);
    }

    @Override
    public RenaiKey key(String key) {
        return kvCommonOperator.key(key, KvType.PriorityQ);
    }

    @Override
    public List<RenaiKey> keys(List<String> keys) {
        return kvCommonOperator.keys(keys, KvType.PriorityQ);
    }

    @Override
    public List<RenaiKey> keys() {
        return kvCommonOperator.keys(KvType.PriorityQ);
    }

    @Override
    public List<RenaiKey> like(String pattern) {
        return kvCommonOperator.like(pattern, KvType.PriorityQ);
    }

    @Override
    public List<RenaiKey> regexp(String pattern) {
        return kvCommonOperator.regexp(pattern, KvType.PriorityQ);
    }

    @Override
    public List<RenaiValue> values(String key) {
        return kvCommonOperator.values(key, KvType.PriorityQ);
    }

    @Override
    public int total() {
        return kvCommonOperator.total(KvType.PriorityQ);
    }

    @Override
    public int size(String key) {
        return kvCommonOperator.size(key, KvType.PriorityQ);
    }

    @Override
    public void expire(String key, int expireSeconds) {
        kvCommonOperator.expire(key, expireSeconds, KvType.PriorityQ);
    }

    @Override
    public boolean exists(String key) {
        return kvCommonOperator.exists(key, KvType.PriorityQ);
    }

    @Override
    public <T> boolean contains(String key, T value) {
        return kvCommonOperator.contains(key, value, KvType.PriorityQ);
    }

    @Override
    public int clear(String key) {
        return kvCommonOperator.clear(key, KvType.PriorityQ);
    }

    @Override
    public <T> boolean remove(String key, T value) {
        return kvCommonOperator.remove(key, value, KvType.PriorityQ);
    }

    @Override
    public <T> int removeValues(String key, Collection<T> values) {
        return kvCommonOperator.removeValues(key, values, KvType.PriorityQ);
    }

    @Override
    public boolean delete(String key) {
        return kvCommonOperator.delete(key, KvType.PriorityQ);
    }

    @Override
    public int delete(Collection<String> keys) {
        return kvCommonOperator.delete(keys, KvType.PriorityQ);
    }

    @Override
    public void flush() {
        kvCommonOperator.flush(KvType.PriorityQ);
    }

    @Override
    public <T> List<Tuple2<T, Long>> members(String key, Class<T> clazz) {
        assert clazz != null;
        List<RenaiValue> renaiValues = kvCommonOperator.values(key, KvType.PriorityQ);
        return renaiValues.stream().map(rv -> new Tuple2<>(RenaiUtil.parse(rv.getV(), clazz), rv.getScore())).toList();
    }

    @Override
    public <T> void add(String key, T value, long score) {
        commonAdd(key, List.of(new Tuple2<>(value, score)), 1);
    }

    @Override
    public <T> void add(String key, List<Tuple2<T, Long>> list) {
        commonAdd(key, list, 1);
    }

    @Override
    public <T> void addIfAbsent(String key, T value, long score) {
        commonAdd(key, List.of(new Tuple2<>(value, score)), 2);
    }

    @Override
    public <T> void addIfAbsent(String key, List<Tuple2<T, Long>> list) {
        commonAdd(key, list, 2);
    }

    @Override
    public <T> void addCover(String key, T value, long score) {
        commonAdd(key, List.of(new Tuple2<>(value, score)), 3);
    }

    @Override
    public <T> void addCover(String key, List<Tuple2<T, Long>> list) {
        commonAdd(key, list, 3);
    }

    private <T> void commonAdd(String key, List<Tuple2<T, Long>> list, int flag) {
        assert StringUtils.isNotBlank(key) && CollectionUtils.isNotEmpty(list)
                && list.stream().noneMatch(t -> t._1 == null || t._2 == null);
        Map<String, Object> params = procedureParams(key);
        params.put(VALUE, RenaiUtil.string(list));
        params.put(FLAG, flag);
        jdbcProxy.call(KV_QUEUE_ADD, params);
    }

    private <T> List<Tuple2<T, Long>> commonGet(String key, int start, int count, Class<T> clazz, boolean rev) {
        assert clazz != null;
        ProcedureResult result = commonRange(key, start, count, rev, 1);
        return result.list(RenaiValue.class).stream()
                .map(v -> new Tuple2<>(RenaiUtil.parse(v.getV(), clazz), v.getScore()))
                .toList();
    }

    @Override
    public <T> List<Tuple2<T, Long>> get(String key, int start, int count, Class<T> clazz) {
        return commonGet(key, start, count, clazz, false);
    }

    @Override
    public <T> List<Tuple2<T, Long>> revGet(String key, int start, int count, Class<T> clazz) {
        return commonGet(key, start, count, clazz, true);
    }

    @Override
    public <T> List<Tuple2<T, Long>> getByScore(String key, long start, long end, Class<T> clazz) {
        assert clazz != null;
        ProcedureResult result = commonRangeScore(key, start, end, 1);
        return result.list(RenaiValue.class).stream()
                .map(v -> new Tuple2<>(RenaiUtil.parse(v.getV(), clazz), v.getScore()))
                .toList();
    }

    private <T> List<Tuple2<T, Long>> commonPop(String key, int start, int count, Class<T> clazz, boolean rev) {
        assert clazz != null;
        ProcedureResult result = commonRange(key, start, count, rev, 3);
        return result.list(RenaiValue.class).stream()
                .map(v -> new Tuple2<>(RenaiUtil.parse(v.getV(), clazz), v.getScore()))
                .toList();
    }

    @Override
    public <T> List<Tuple2<T, Long>> pop(String key, int start, int count, Class<T> clazz) {
        return commonPop(key, start, count, clazz, false);
    }

    @Override
    public <T> List<Tuple2<T, Long>> revPop(String key, int start, int count, Class<T> clazz) {
        return commonPop(key, start, count, clazz, true);
    }

    @Override
    public <T> List<Tuple2<T, Long>> popByScore(String key, long start, long end, Class<T> clazz) {
        assert clazz != null;
        ProcedureResult result = commonRangeScore(key, start, end, 3);
        return result.list(RenaiValue.class).stream()
                .map(v -> new Tuple2<>(RenaiUtil.parse(v.getV(), clazz), v.getScore()))
                .toList();
    }

    private int commonDelete(String key, int start, int count, boolean rev) {
        ProcedureResult result = commonRange(key, start, count, rev, 2);
        return result.updated();
    }

    @Override
    public int delete(String key, int start, int count) {
        return commonDelete(key, start, count, false);
    }

    @Override
    public int revDelete(String key, int start, int count) {
        return commonDelete(key, start, count, true);
    }

    @Override
    public int deleteByScore(String key, long start, long end) {
        ProcedureResult result = commonRangeScore(key, start, end, 2);
        return result.updated();
    }

    private ProcedureResult commonRange(String key, int start, int count, boolean rev, int flag) {
        assert StringUtils.isNotBlank(key) && start >= 0 && count > 0;
        Map<String, Object> params = procedureParams(key);
        params.put(START, start);
        params.put(SIZE, count);
        params.put(REV, rev ? 1 : -1);
        params.put(FLAG, flag);
        return jdbcProxy.call(KV_QUEUE_RANGE, params);
    }

    public ProcedureResult commonRangeScore(String key, long start, long end, int flag) {
        assert StringUtils.isNotBlank(key) && start <= end;
        Map<String, Object> params = procedureParams(key);
        params.put(START, start);
        params.put(END, end);
        params.put(FLAG, flag);
        return jdbcProxy.call(KV_QUEUE_RANGE_SCORE, params);
    }

    @Override
    public <T> List<Long> score(String key, T value) {
        assert StringUtils.isNotBlank(key) && value != null;
        String sql = "SELECT score FROM renai_kv_queue_value_valid WHERE k = ? AND v = ? ORDER BY score";
        return jdbcProxy.queryForList(sql, Long.class, key, RenaiUtil.string(value));
    }

    @Override
    public <T> Long minScore(String key, T value) {
        return score(key, value).stream().min(Comparator.comparing(s -> s)).orElse(null);
    }

    @Override
    public <T> Long maxScore(String key, T value) {
        return score(key, value).stream().max(Comparator.comparing(s -> s)).orElse(null);
    }

    private <T> List<Integer> commonRank(String key, T value, boolean rev) {
        assert StringUtils.isNotBlank(key) && value != null;
        Map<String, Object> params = procedureParams(key);
        params.put(VALUE, RenaiUtil.string(value));
        params.put(REV, rev ? 1 : -1);
        ProcedureResult result = jdbcProxy.call(KV_QUEUE_RANK, params);
        return result.list(ValueResult.class).stream().map(ValueResult::getIv).toList();
    }

    @Override
    public <T> List<Integer> rank(String key, T value) {
        return commonRank(key, value, false);
    }

    @Override
    public <T> Integer highestRank(String key, T value) {
        return commonRank(key, value, false).stream().min(Comparator.comparing(s -> s)).orElse(null);
    }

    @Override
    public <T> Integer lowestRank(String key, T value) {
        return commonRank(key, value, false).stream().max(Comparator.comparing(s -> s)).orElse(null);
    }

    @Override
    public <T> List<Integer> revRank(String key, T value) {
        return commonRank(key, value, true);
    }

    @Override
    public <T> Integer revHighestRank(String key, T value) {
        return commonRank(key, value, true).stream().min(Comparator.comparing(s -> s)).orElse(null);
    }

    @Override
    public <T> Integer revLowestRank(String key, T value) {
        return commonRank(key, value, true).stream().max(Comparator.comparing(s -> s)).orElse(null);
    }

    @Override
    public <T> List<Long> incr(String key, T value, long score) {
        assert StringUtils.isNotBlank(key) && value != null;
        Map<String, Object> params = procedureParams(key);
        params.put(VALUE, RenaiUtil.string(value));
        params.put(SCORE, score);
        ProcedureResult result = jdbcProxy.call(KV_QUEUE_INCR, params);
        return result.list(ValueResult.class).stream().map(ValueResult::getLv).toList();
    }

}
